package se.sics.drsm.components;

/**
 * This file is part of the Kompics P2P Framework.
 *
 * Copyright (C) 2009 Swedish Institute of Computer Science (SICS)
 * Copyright (C) 2009 Royal Institute of Technology (KTH)
 *
 * Kompics is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import se.sics.drsm.inits.DRSMIterativeRouterInit;
import se.sics.drsm.model.DRSMAddress;
import se.sics.drsm.model.DRSMFingerTable;
import se.sics.drsm.model.LookupInfo;
import se.sics.drsm.overlay_address.key.NumericRingKey;
import se.sics.drsm.overlay_address.key.RingKey.IntervalBounds;
import se.sics.drsm.ports.drsm.DRSMPeriodicStabilizationPort;
import se.sics.drsm.ports.drsm.DRSMRouter;
import se.sics.drsm.ports.drsm.events.*;
import se.sics.drsm.ports.statemachine.StateMachinePort;
import se.sics.drsm.ports.statemachine.events.UpdatePeers;
import se.sics.drsm.util.CollectionUtil;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.CancelTimeout;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;

import java.math.BigInteger;
import java.util.*;

/**
 * The <code>ChordIterativeRouter</code> class.
 *
 * @author Cosmin Arad <cosmin@sics.se>
 * @author Shadid Chowdhury <msrch@kth.se>
 */
public final class DRSMIterativeRouter extends ComponentDefinition {

    Negative<DRSMRouter> router = negative(DRSMRouter.class);
    Positive<DRSMPeriodicStabilizationPort> ps = positive(DRSMPeriodicStabilizationPort.class);
    Positive<StateMachinePort> stateMachinePort = positive(StateMachinePort.class);
    Positive<Network> network = positive(Network.class);
    Positive<Timer> timer = positive(Timer.class);

    private Logger logger;
    private int log2RingSize;
    private long fingerStabilizationPeriod;
    private long rpcTimeout;
    private BigInteger ringSize;
    private DRSMAddress self;
    private List<DRSMAddress> peers, predecessors, successors;
    private DRSMFingerTable fingerTable;
    private HashMap<UUID, LookupInfo> outstandingLookups;
    private final HashSet<UUID> outstandingTimeouts;
    private DRSMIterativeRouter thisRouter;

    public DRSMIterativeRouter() {
        thisRouter = this;
        outstandingLookups = new HashMap<UUID, LookupInfo>();
        outstandingTimeouts = new HashSet<UUID>();

        subscribe(handleInit, control);

        subscribe(handleFindSuccessorRequest, network);
        subscribe(handleFindSuccessorResponse, network);
        subscribe(handleGetFingerTableRequest, network);
        subscribe(handleGetFingerTableResponse, network);

        subscribe(handleDRSMLookupRequest, router);

        subscribe(handleNewSuccessors, ps);
        subscribe(handleNewPredecessor, ps);

        subscribe(handleUpdatePeers, stateMachinePort);

        subscribe(handleRpcTimeout, timer);
        subscribe(handleFixFingersTimeout, timer);
    }

    Handler<DRSMIterativeRouterInit> handleInit = new Handler<DRSMIterativeRouterInit>() {
        public void handle(DRSMIterativeRouterInit init) {
            fingerStabilizationPeriod = init.getFingerStabilizationPeriod();
            log2RingSize = init.getLog2RingSize();
            rpcTimeout = init.getRpcTimeout();

            ringSize = new BigInteger("2").pow(log2RingSize);
            self = init.getSelf();
            peers = new ArrayList<DRSMAddress>();
            peers.add(self);

            fingerTable = new DRSMFingerTable(log2RingSize, self, thisRouter);

            fingerTableChanged();

            logger = LoggerFactory.getLogger(getClass().getName() + "@"
                    + self.getKey().getId());
        }
    };

    private Handler<DRSMNewSuccessors> handleNewSuccessors = new Handler<DRSMNewSuccessors>() {
        public void handle(DRSMNewSuccessors event) {
            logger.debug("NEW_SUCC {}", event.getSuccessorListView());

            if (successors == null) {
                // if this is the first time we get this event, we initialize
                // the periodic FixFingers timer
                SchedulePeriodicTimeout spt = new SchedulePeriodicTimeout(
                        fingerStabilizationPeriod, fingerStabilizationPeriod);
                spt.setTimeoutEvent(new FixFingersTimeout(spt));
                trigger(spt, timer);

                // FIXME: successor list view does it contain next successor or
                // randomly many
                // successor = event.getSuccessorListView().get(0);
                successors = event.getSuccessorListView();

                if (!successors.equals(self)) { //FIXME
                    // ask successor for its finger table
                    DRSMGetFingerTableRequest request = new DRSMGetFingerTableRequest(
                            self, CollectionUtil.getFirstElem(successors));
                    trigger(request, network);
                }
            }

            // FIXME
            // successor = event.getSuccessorListView().get(0);
            successors = event.getSuccessorListView();

            boolean changed = false;
            // try to use the successors
            for (DRSMAddress address : event.getSuccessorListView()) {
                if (fingerTable.learnedAboutPeer(address, false, false)) {
                    changed = true;
                }
            }
            if (changed) {
                fingerTableChanged();
            }
        }
    };

    private Handler<DRSMNewPredecessors> handleNewPredecessor = new Handler<DRSMNewPredecessors>() {
        public void handle(DRSMNewPredecessors event) {
            logger.debug("NEW_PRED {}", event.getPredecessorPeers());

            predecessors = event.getPredecessorPeers();
        }
    };

    private Handler<DRSMLookupRequest> handleDRSMLookupRequest = new Handler<DRSMLookupRequest>() {
        public void handle(DRSMLookupRequest event) {
            logger.debug("handling DRSMLookupRequest key {} ", event.getKey());
            DRSMLookupResponse response = doLookup(event, false);
            if (response != null) {
                trigger(response, router);
            }
        }
    };

    private final DRSMLookupResponse doLookup(DRSMLookupRequest event,
                                              boolean maintenance) {
        logger.debug("doLookup({})", event.getKey());

        NumericRingKey key = event.getKey();
        DRSMAddress firstPeer = event.getFirstPeer();
        boolean isJoinRequest = event.isJoinRequest();

        // special case for join ring, first peer is one of the node that we learned from bootstrap server
        if (firstPeer != null) {

            logger.debug("doLookup through FIRST_PEER {}", firstPeer);

            LookupInfo lookupInfo = new LookupInfo(event);
            lookupInfo.initiatedNow();

            ScheduleTimeout st = new ScheduleTimeout(rpcTimeout);
            st.setTimeoutEvent(new RpcTimeout(st, event, firstPeer, maintenance));
            UUID timeoutId = st.getTimeoutEvent().getTimeoutId();
            outstandingLookups.put(timeoutId, lookupInfo);
            outstandingTimeouts.add(timeoutId);
            trigger(st, timer);

            FindSuccessorsRequest request = new FindSuccessorsRequest(key,
                    timeoutId, self, firstPeer, maintenance);

            trigger(request, network);

            // we try to use the hinted first peer as a possible better
            // finger
            fingerTable.learnedAboutFreshPeer(firstPeer);
            return null;
        }

        // special case for when we are alone in the ring
        if (successors == null || successors.equals(self)) {
            logger.debug("Only one group in the ring is {}", firstPeer);
            // to avoid an infinite loop, we return ourselves
            return new DRSMLookupResponse(event, key, peers,
                    event.getAttachment(), new LookupInfo(event), isJoinRequest);
        }

        // normal case
        if (key.equals(self.getKey())) {
            // we are responsible
            return new DRSMLookupResponse(event, key, peers,
                    event.getAttachment(), new LookupInfo(event), isJoinRequest);
        } else if (predecessors != null
                && key.belongsTo(CollectionUtil.getFirstElem(predecessors).getKey(), self.getKey(),
                IntervalBounds.OPEN_CLOSED, ringSize)) {
            // we are responsible
            return new DRSMLookupResponse(event, key, peers,
                    event.getAttachment(), new LookupInfo(event), isJoinRequest);
        } else if (key.belongsTo(self.getKey(), CollectionUtil.getFirstElem(successors).getKey(), //FIXME: verify this works
                IntervalBounds.OPEN_CLOSED, ringSize)) {
            // our successor is responsible for the looked up key
            return new DRSMLookupResponse(event, key, successors,
                    event.getAttachment(), new LookupInfo(event), isJoinRequest);
        } else {
            // some other peer is responsible for the looked up key
            LookupInfo lookupInfo = new LookupInfo(event);
            lookupInfo.initiatedNow();

            List<DRSMAddress> closest = fingerTable.closestPreceedingPeers(key);

            if (closest == null || closest.contains(self)) {
                // we found no closest peer so the lookup should fail
                return new DRSMLookupResponse(event, key,
                        event.getAttachment(), null, isJoinRequest);
            }

            ScheduleTimeout st = new ScheduleTimeout(rpcTimeout);
            st.setTimeoutEvent(new RpcTimeout(st, event, CollectionUtil.getFirstElem(closest), maintenance));
            UUID timeoutId = st.getTimeoutEvent().getTimeoutId();
            outstandingLookups.put(timeoutId, lookupInfo);
            outstandingTimeouts.add(timeoutId);
            trigger(st, timer);

            FindSuccessorsRequest request = new FindSuccessorsRequest(key,
                    timeoutId, self, CollectionUtil.getFirstElem(closest), maintenance);

            trigger(request, network);
            return null;
        }
    }

    private Handler<FindSuccessorsRequest> handleFindSuccessorRequest = new Handler<FindSuccessorsRequest>() {
        public void handle(FindSuccessorsRequest event) {
            logger.debug(
                    "FIND_SUCC_REQ@{} for key {},succ={},pred={}",
                    new Object[]{self, event.getKey(), successors, predecessors});

            NumericRingKey key = event.getKey();

            if (key.equals(self.getKey())) {
                logger.debug(
                        "FIND_SUCC_REQ{} responsible self {}", event.getKey(), self);
                // return ourselves as the real responsible
                FindSuccessorsResponse response = new FindSuccessorsResponse(
                        peers, event.getLookupId(), false, self,
                        event.getDRSMSource(), event.isMaintenance());
                trigger(response, network);
            } else if (predecessors != null && predecessors.size() > 0
                    && key.belongsTo(CollectionUtil.getFirstElem(predecessors).getKey(), self.getKey(),
                    IntervalBounds.OPEN_CLOSED, ringSize)) {
                logger.debug(
                        "FIND_SUCC_REQ{} responsible self {}", event.getKey(), self);
                // return ourselves as the real responsible
                FindSuccessorsResponse response = new FindSuccessorsResponse(
                        peers, event.getLookupId(), false, self,
                        event.getDRSMSource(), event.isMaintenance());
                trigger(response, network);
            } else if (successors != null && successors.size() > 0) {
                if (key.belongsTo(self.getKey(), CollectionUtil.getFirstElem(successors).getKey(),
                        IntervalBounds.OPEN_CLOSED, ringSize)) {
                    logger.debug(
                            "FIND_SUCC_REQ{} responsible our successors {}", event.getKey(), successors);
                    // return my successor as the real successor
                    FindSuccessorsResponse response = new FindSuccessorsResponse(
                            successors, event.getLookupId(), false, self,
                            event.getDRSMSource(), event.isMaintenance());
                    trigger(response, network);
                } else {
                    logger.debug(
                            "FIND_SUCC_REQ{} responsible closest preceeding peer", event.getKey());
                    // return an indirection to my closest preceding finger
                    List<DRSMAddress> closest = fingerTable
                            .closestPreceedingPeers(key);

                    if (closest != null && closest.size() > 0 && !closest.contains(self)) {
                        FindSuccessorsResponse response = new FindSuccessorsResponse(
                                closest, event.getLookupId(), true, self,
                                event.getDRSMSource(), event.isMaintenance());
                        trigger(response, network);
                    }
                    // else we found no closest peer so the lookup should fail
                }
            }

            // we try to use the requester as a possible better finger
            fingerTable.learnedAboutFreshPeer(event.getDRSMSource());
        }
    };

    private Handler<FindSuccessorsResponse> handleFindSuccessorResponse = new Handler<FindSuccessorsResponse>() {
        public void handle(FindSuccessorsResponse event) {
            List<DRSMAddress> successors = event.getSuccessor();
            UUID lookupId = event.getLookupId();
            if (outstandingTimeouts.contains(lookupId)) {
                // we got the response before the RpcTimeout, so we cancel the
                // timer
                trigger(new CancelTimeout(lookupId), timer);
                outstandingTimeouts.remove(lookupId);
            } else {
                // we got the response too late so we just give up since we have
                // already triggered a LookupFailed event
                return;
            }

            LookupInfo lookupInfo = outstandingLookups.remove(lookupId);
            DRSMLookupRequest lookupRequest = lookupInfo.getRequest();

            logger.debug("FIND_SUCC_RESP NH({}) is {}. {}",
                    new Object[]{lookupRequest.getKey(),
                            event.getSuccessor(), event.isNextHop()});

            if (event.isNextHop()) {
                // we got an indirection
                DRSMAddress nextHop = CollectionUtil.getRandomElem(successors);

                ScheduleTimeout st = new ScheduleTimeout(rpcTimeout);
                st.setTimeoutEvent(new RpcTimeout(st, lookupRequest, nextHop,
                        event.isMaintenance()));
                UUID timeoutId = st.getTimeoutEvent().getTimeoutId();
                lookupInfo.appendHop(event.getDRSMSource());
                outstandingLookups.put(timeoutId, lookupInfo);
                outstandingTimeouts.add(timeoutId);
                trigger(st, timer);

                FindSuccessorsRequest request = new FindSuccessorsRequest(
                        lookupRequest.getKey(), timeoutId, self, nextHop,
                        event.isMaintenance());

                // send find successor request
                trigger(request, network);
            } else {
                // we got the real responsible
                lookupInfo.appendHop(event.getDRSMSource());
                lookupInfo.completedNow();
                DRSMLookupResponse response = new DRSMLookupResponse(
                        lookupRequest, lookupRequest.getKey(),
                        successors, lookupRequest.getAttachment(),
                        lookupInfo, lookupRequest.isJoinRequest());

                if (event.isMaintenance()) {
                    List<DRSMAddress> fingerPeers = response.getResponsible();
                    int fingerIndex = (Integer) response.getAttachment();
                    fingerTable.fingerFixed(fingerIndex, fingerPeers);
                } else {
                    // System.err.println(response.getLookupInfo());
                    trigger(response, router);
                }
            }

            // we try to use the responsible or the next hop as a possible
            // better finger
            fingerTable.learnedAboutPeer(CollectionUtil.getFirstElem(successors));
        }
    };

    private Handler<UpdatePeers> handleUpdatePeers = new Handler<UpdatePeers>() {
        public void handle(UpdatePeers event) {
            logger.debug("handleUpdatePeers {}", event.toString());
            peers = event.getPeerList();
        }
    };

    private Handler<RpcTimeout> handleRpcTimeout = new Handler<RpcTimeout>() {
        public void handle(RpcTimeout event) {
            logger.debug("RPC_TIMEOUT");

            UUID timeoutId = event.getTimeoutId();

            if (outstandingTimeouts.contains(timeoutId)) {
                outstandingTimeouts.remove(timeoutId);

                // we got an RPC timeout before the RPC response so we have to
                // return a ChordLookupFailed event and to mark the finger as
                // failed
                DRSMLookupRequest lookupRequest = outstandingLookups.remove(
                        timeoutId).getRequest();
                DRSMLookupResponse failed = new DRSMLookupResponse(
                        lookupRequest, lookupRequest.getKey(),
                        lookupRequest.getAttachment(), event.getPeer(), lookupRequest.isJoinRequest());

                if (!event.isMaintenance()) {
                    trigger(failed, router);
                }

                // mark the slow/failed peer as suspected
                fingerTable.fingerSuspected(event.getPeer());
            }
            // else we got the RPC response before the timeout so everything is
            // OK.
        }
    };

    private Handler<FixFingersTimeout> handleFixFingersTimeout = new Handler<FixFingersTimeout>() {
        public void handle(FixFingersTimeout event) {
            logger.debug("FIX_FINGER");

            int nextFingerToFix = fingerTable.nextFingerToFix();

            // actually fix the finger now
            // finger[next] = findSuccessor(n plus 2^(next-1))
            NumericRingKey fingerBegin = fingerTable
                    .getFingerBegin(nextFingerToFix);

            DRSMLookupRequest request = new DRSMLookupRequest(fingerBegin,
                    nextFingerToFix, false);

            // here I need to use myself
            DRSMLookupResponse response = doLookup(request, true);
            if (response != null) {
                if (response.getStatus() == DRSMLookupResponse.DRSMLookupStatus.FAILURE) {
                    logger.debug("CHORD_LOOKUP_FAILED");
                    DRSMAddress fingerPeer = response.getSuspected();
                    fingerTable.fingerSuspected(fingerPeer);
                } else {
                    logger.debug("CHORD_LOOKUP_RESP");

                    List<DRSMAddress> fingerPeer = response.getResponsible();
                    int fingerIndex = (Integer) response.getAttachment();
                    fingerTable.fingerFixed(fingerIndex, fingerPeer);
                }
            }
        }
    };

    private Handler<GetFingerTableRequest> handleGetFingerTableRequest = new Handler<GetFingerTableRequest>() {
        public void handle(GetFingerTableRequest event) {
            logger.debug("GET_FINGER_TABLE_REQ");

            GetFingerTableResponse response = new GetFingerTableResponse(
                    fingerTable.getView(), self, event.getDRSMSource());
            trigger(response, network);
        }
    };

    private Handler<GetFingerTableResponse> handleGetFingerTableResponse = new Handler<GetFingerTableResponse>() {
        public void handle(GetFingerTableResponse event) {
            logger.debug("GET_FINGER_TABLE_RESP");

            // we try to populate our finger table with our successor's finger
            // table
            boolean changed = false;
            // try to use the successors
            Map<Integer, List<DRSMAddress>> fingerMap = event.getFingerTable().fingerMap;
            List<DRSMAddress> addressList;
            for (Integer index : fingerMap.keySet()) {
                addressList = fingerMap.get(index);
                if (fingerTable.learnedAboutPeers(addressList, false, false)) {
                    changed = true;
                }
            }
            if (changed) {
                fingerTableChanged();
            }
            // trigger(new JoinRingCompleted(localPeer),
            // notificationChannel);
        }
    };

    public final void fingerTableChanged() {
        trigger(new CreateDRSMFingerTable(fingerTable.getView()), router);
    }
}